package org.devgframwork.datasync.core.support.executors;

import com.alibaba.fastjson.JSONObject;
import org.devgframwork.datasync.core.entity.CoConfig;
import org.devgframwork.datasync.core.entity.TargetConfig;
import org.devgframwork.datasync.core.model.ExecutorTypeEnum;
import org.devgframwork.datasync.core.model.ResultModel;
import org.devgframwork.datasync.core.support.executors.target.TargetEs56Executor;
import org.devgframwork.datasync.core.support.executors.target.TargetSqlExecutor;
import org.devgframwork.datasync.core.support.service.ScheduleConfigService;
import org.devgframwork.datasync.core.utils.CommonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

public class ExecutorCore implements Callable<ResultModel> {

    private static Logger log = LoggerFactory.getLogger(ExecutorCore.class);

    private int page;
    private int batchSize;
    private String sqlText;
    private JdbcTemplate sourceJdbcTemplate;
    private ScheduleConfigService scheduleConfigService;
    private CoConfig coConfig;
    private List<TargetConfig> targetConfigList;
    private Map<String, Object> paramMap;


    public ExecutorCore(int page, int batchSize, String sqlText, JdbcTemplate sourceJdbcTemplate,
                        ScheduleConfigService scheduleConfigService, CoConfig coConfig,
                        List<TargetConfig> targetConfigList, Map<String, Object> paramMap) {
        this.page = page;
        this.batchSize = batchSize;
        this.sqlText = sqlText;
        this.sourceJdbcTemplate = sourceJdbcTemplate;
        this.scheduleConfigService = scheduleConfigService;
        this.coConfig = coConfig;
        this.targetConfigList = targetConfigList;
        this.paramMap = paramMap;
    }


    @Override
    public ResultModel call() {
        ThreadPoolExecutor threadPoolExecutor = null;
        try {
            String executeSql = null;
            int startPosition = 0;
            if (CommonUtil.eq(coConfig.getSourceType(), ExecutorTypeEnum.jdbc_mysql.getTypeName())) {
                startPosition = page * batchSize;
                executeSql = sqlText + " limit " + startPosition + "," + batchSize;
            }
            if (CommonUtil.isNullOrEmpty(executeSql)) {
                return ResultModel.newFail("executeSql cannot be null");
            }
            log.info("executeName:{},executeSql:{}", coConfig.getExecuteName(), executeSql);
            List<Map<String, Object>> resultList = sourceJdbcTemplate.queryForList(executeSql);
            if (null == resultList || resultList.size() <= 0) {
                return ResultModel.newSuccess();
            }
            threadPoolExecutor = new ThreadPoolExecutor(
                    4,
                    4,
                    30L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<Runnable>(10),
                    new ThreadFactory() {
                        @Override
                        public Thread newThread(Runnable r) {
                            return new Thread(r, "data-sync, admin JdbcToJdbcExecutor-threadPoolExecutorThreadPool-" + r.hashCode());
                        }
                    },
                    new RejectedExecutionHandler() {
                        @Override
                        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                            r.run();
                            log.warn(">>>>>>>>>>> data-sync, registry or remove too fast, match threadpool rejected handler(run now).");
                        }
                    });
            Map<String, Future<ResultModel>> targetFutureMap = new HashMap<>();
            for (int i = 0; i < targetConfigList.size(); i++) {
                TargetConfig targetConfig = targetConfigList.get(i);
                String key = targetConfig.getTargetKey() + "_" + i;
                Future<ResultModel> future = null;
                if (CommonUtil.eq(targetConfig.getTargetType(), ExecutorTypeEnum.jdbc_mysql.getTypeName())) {
                    TargetSqlExecutor targetSqlExecutor = new TargetSqlExecutor(resultList, targetConfig);
                    future = threadPoolExecutor.submit(targetSqlExecutor);
                }
                if (CommonUtil.eq(targetConfig.getTargetType(), ExecutorTypeEnum.es56.getTypeName())) {
                    TargetEs56Executor targetEs56Executor = new TargetEs56Executor(resultList, targetConfig, paramMap);
                    future = threadPoolExecutor.submit(targetEs56Executor);
                }
                if (null != future) {
                    targetFutureMap.put(key, future);
                }
            }
            for (String key : targetFutureMap.keySet()) {
                Future<ResultModel> future = targetFutureMap.get(key);
                ResultModel resultModel = future.get();
                log.info("JdbcMySqlExecutor ExecutorCore key:{},resultModel:{}", key, JSONObject.toJSONString(resultModel));
            }
            if (null != scheduleConfigService) {
                Map<String, Object> updateSyncMap = new HashMap<>();
                updateSyncMap.put("taskName", coConfig.getExecuteName());
                updateSyncMap.put("syncNum", batchSize);
                scheduleConfigService.updateSyncNumById(updateSyncMap);
            }
            log.info("finish sync taskName:" + coConfig.getExecuteName() + ",from:" + startPosition + ",batchSize:" + batchSize);
            return ResultModel.newSuccess();
        } catch (Exception ex) {
            log.error("JdbcToJdbcExecutor error", ex.getMessage());
            ex.printStackTrace();
            return ResultModel.newFail(ex.getMessage());
        } finally {
            if (null != threadPoolExecutor) {
                threadPoolExecutor.shutdown();
            }
        }
    }
}
